package xkafka

import (
	"context"
	"time"

	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/plain"
)

type ReaderConfig struct {
	Username    string
	Password    string
	Brokers     []string
	GroupID     string
	Topic       string
	Logger      Logger
	Partition   int
	OffsetReset bool
}

//var reader *kafka.Reader

func NewConsumer(ctx context.Context, config ReaderConfig, f func(message kafka.Message) (err error)) (err error) {
	if config.Logger == nil {
		logger := &DefaultLogger{}
		logger.SetPrefix(ReaderLogPrefix)
		config.Logger = logger
	}
	offset := kafka.LastOffset
	if config.OffsetReset {
		offset = kafka.FirstOffset
	}

	readerConfig := kafka.ReaderConfig{
		Brokers:     config.Brokers,
		Partition:   config.Partition,
		GroupID:     config.GroupID,
		Topic:       config.Topic,
		Logger:      kafka.LoggerFunc(config.Logger.Infof),
		ErrorLogger: kafka.LoggerFunc(config.Logger.Errorf),
	}

	// 处理消费者组
	if len(config.GroupID) > 0 {
		readerConfig.WatchPartitionChanges = true
		readerConfig.PartitionWatchInterval = 3 * time.Second
		readerConfig.StartOffset = offset
	}

	// 验证
	if len(config.Username) > 0 {
		dialer := &kafka.Dialer{
			SASLMechanism: plain.Mechanism{
				Username: config.Username,
				Password: config.Password,
			},
		}
		readerConfig.Dialer = dialer
	}

	err = readerConfig.Validate()
	if err != nil {
		config.Logger.Errorf("ReaderConfig err: " + err.Error())
		return
	}

	reader := kafka.NewReader(readerConfig)
	// 关闭
	defer func() {
		err = reader.Close()
		if err != nil {
			config.Logger.Errorf("Close err: " + err.Error())
		}
	}()

	if len(config.GroupID) == 0 {
		err = reader.SetOffset(offset)
		if err != nil {
			config.Logger.Errorf("SetOffset err: " + err.Error())
			return
		}
	}

	for {
		message, err := reader.FetchMessage(ctx)
		if err != nil {
			config.Logger.Errorf("FetchMessage err: " + err.Error())
			break
		}
		err = f(message)
		if err != nil {
			config.Logger.Errorf("FuncMessage err: " + err.Error())
		}

		if len(reader.Config().GroupID) > 0 {
			err = reader.CommitMessages(ctx, message)
			if err != nil {
				config.Logger.Errorf("CommitMessages err: " + err.Error())
			}
		}

	}

	return
}
